package com.example.mqtt_mode.mqtt.config;

import cn.hutool.json.JSONUtil;
import com.example.mqtt_mode.mqtt.generator.domain.Test1;
import com.example.mqtt_mode.mqtt.generator.service.Test1Service;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author IT空门_门主
 * @date 2024/1/4
 * MQTT的消息回调
 */
@Slf4j
@Component
@SuppressWarnings("all")
public class MqttMessageCallback  implements MqttCallback {

    /**
     * 连接丢失后的处理
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.info(throwable.getMessage());
        log.info("连接断开，正在重新连接...:{}", throwable.getMessage());
        MqttConfig mqttConfig = new MqttConfig();
        mqttConfig.startReconnect();
        log.info("重新连接成功");

    }

    /**
     * 收到消息后的处理
     * @param s
     * @param mqttMessage
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
           log.info("收到消息主题:{},消息内容:{},消息的推送级别次数:{}"
                   ,topic,new String(mqttMessage.getPayload()),mqttMessage.getQos());
        ExecutorService executor = Executors.newFixedThreadPool(5);
            executor.execute(new Runnable() {
                public void run() {
                    log.info("线程开始执行");
                    saveDate(mqttMessage);
                    log.info("线程执行结束");
                }
            });
        executor.shutdown(); // 关闭线程池
    }

    /**
     *  消息发送后的处理
     * @param iMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("消息发送后的处理");

    }
    /**
     * 处理要保存的数据
     */
    public  void saveDate(MqttMessage mqttMessage){
        log.info("线程执行中");
        log.info("-----处理要保存的数据-----");
        if (new String(mqttMessage.getPayload()).equals("offline")){
            log.info("消息内容为空");
            return ;
        }
        Map bean = JSONUtil.toBean(new String(mqttMessage.getPayload()), Map.class);
        Test1 test1 = new Test1();
        test1.setUserName(bean.get("userName").toString());
        test1.setAge(Integer.parseInt(bean.get("age").toString()));
        log.info("结果:{}", bean.get("userName"));
        log.info("结果:{}", bean.get("age"));
        //手动注入 Bean
        Test1Service test1Mapper = SpringJobBeanFactory.getBean(Test1Service.class);
        test1Mapper.save(test1);

    }

}
